package cn.xinfei.xdecision.thread;

import io.micrometer.core.instrument.ImmutableTag;
import io.micrometer.core.instrument.Metrics;
import io.micrometer.core.instrument.Tag;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;

import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

@Slf4j
public class CustomerThreadPoolExecutor extends ThreadPoolExecutor {

    public static Map<String, CustomerThreadPoolExecutor> customerThreadPoolExecutorMap = new ConcurrentHashMap<String, CustomerThreadPoolExecutor>();
    public static Map<String, ThreadPoolExecutorConfig> threadPoolExecutorConfigMap = new ConcurrentHashMap<String, ThreadPoolExecutorConfig>();
    /**
     * 保存任务开始执行的时间，当任务结束时，用任务结束时间减去开始时间计算任务执行时间
     */
    private ConcurrentHashMap<String, Date> startTimes;
    /**
     * 线程池名称，一般以业务名称命名，方便区分
     */
    private String poolName;

//    private Semaphore semaphore;


    public CustomerThreadPoolExecutor(ThreadPoolExecutorConfig config) {
        this(config, new LinkedBlockingQueue<>(config.getMaxQueueSize()), CustomerThreadPoolExecutorBuilder.handler, new CustomerThreadFactory(config.getName()));
    }

    public CustomerThreadPoolExecutor(String poolName) {
        super(0, Integer.MAX_VALUE,
                60L, TimeUnit.SECONDS,
                new SynchronousQueue<>(),
                new CustomerThreadFactory(poolName));
        customerThreadPoolExecutorMap.put(poolName, this);
        this.startTimes = new ConcurrentHashMap<>();
        this.poolName = poolName;
        //预热
        this.prestartAllCoreThreads();
    }

    public CustomerThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        //预热
        this.prestartAllCoreThreads();
    }

    public CustomerThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        this.startTimes = new ConcurrentHashMap<>();
        this.poolName = "Thread";
        //预热
        this.prestartAllCoreThreads();
    }

    public CustomerThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        //预热
        customerThreadPoolExecutorMap.put(poolName, this);
        this.startTimes = new ConcurrentHashMap<>();
        this.poolName = poolName;
        this.prestartAllCoreThreads();
    }

    /**
     * 调用父类的构造方法，并初始化HashMap和线程池名称
     *
     * @param workQueue     保存被提交任务的队列
     * @param threadFactory 线程工厂
     */
    public CustomerThreadPoolExecutor(ThreadPoolExecutorConfig config, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler, ThreadFactory threadFactory) {
        super(config.getCorePoolSize(), config.getMaximumPoolSize(), config.getKeepAliveTime(), TimeUnit.SECONDS, workQueue, threadFactory, handler);
        customerThreadPoolExecutorMap.put(config.getName(), this);
        threadPoolExecutorConfigMap.put(config.getName(), config);
        this.startTimes = new ConcurrentHashMap<>();
        this.poolName = config.getName();
        //预热
        this.prestartAllCoreThreads();
//        if (config.isSemaphore()) {
//            this.semaphore = new Semaphore(config.getCorePoolSize());
//        }
    }

    public CustomerThreadPoolExecutor(String poolName, ThreadFactory threadFactory) {
        super(10, Integer.MAX_VALUE,
                60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(),
                threadFactory);
        //不优雅停机
//        customerThreadPoolExecutorMap.put(poolName, this);
        this.startTimes = new ConcurrentHashMap<>();
        this.poolName = poolName;

    }

    /**
     * 任务执行之前，记录任务开始时间
     */
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        startTimes.put(String.valueOf(r.hashCode()), new Date());
    }

    private Map<String, String> getContextForTask() {
        return MDC.getCopyOfContextMap();
    }

    @Override
    public void execute(@NonNull Runnable command) {
        super.execute(wrap(command, getContextForTask()));
    }


    /**
     * @param task
     * @param context
     * @param <T>
     * @return
     */
    private static <T> Callable<T> wrap(final Callable<T> task, final Map<String, String> context) {
        return () -> {
            if (!context.isEmpty()) {
                MDC.setContextMap(context);
            }

            try {
                return task.call();
            } finally {
                if (!context.isEmpty()) {
                    MDC.clear();
                }
            }
        };
    }

    private static Runnable wrap(final Runnable runnable, final Map<String, String> context) {
        return () -> {
            if (!context.isEmpty()) {
                MDC.setContextMap(context);
            }

            try {
                runnable.run();
            } finally {
                if (!context.isEmpty()) {
                    MDC.clear();
                }
            }
        };
    }

    @NonNull
    @Override
    public <T> Future<T> submit(@NonNull Callable<T> task) {
//        if (null != semaphore) {
//            try {
//                semaphore.acquire();
//            } catch (InterruptedException e) {
//                log.warn("{} 信号量阻塞被打断", poolName, e);
//            }
//        }
        return super.submit(wrap(task, getContextForTask()));
    }

    @Override
    public Future<?> submit(Runnable task) {
//        if (null != semaphore) {
//            try {
//                semaphore.acquire();
//            } catch (InterruptedException e) {
//                log.warn("{} 信号量阻塞被打断", poolName, e);
//            }
//        }
        return super.submit(wrap(task, getContextForTask()));
    }

    /**
     * 任务执行之后，计算任务结束时间
     */
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
//        if (null != semaphore) {
//            semaphore.release();
//            log.debug("{}-monitor 信号量释放成功", poolName);
//        }
        Date startDate = startTimes.remove(String.valueOf(r.hashCode()));
        Date finishDate = new Date();
        long diff = finishDate.getTime() - startDate.getTime();
        // 统计任务耗时、初始线程数、核心线程数、正在执行的任务数量、
        // 已完成任务数量、任务总数、队列里缓存的任务数量、池中存在的最大线程数、
        // 最大允许的线程数、线程空闲时间、线程池是否关闭、线程池是否终止
        log.info("ThreadMonitor:{} : " +
                        "Duration: {} ms, PoolSize: {}, CorePoolSize: {}, Active: {}, " +
                        "Completed: {}, Task: {}, Queue: {}, LargestPoolSize: {}, " +
                        "MaximumPoolSize: {},  KeepAliveTime: {}, isShutdown: {}, isTerminated: {}",
                this.poolName,
                diff, this.getPoolSize(), this.getCorePoolSize(), this.getActiveCount(),
                this.getCompletedTaskCount(), this.getTaskCount(), this.getQueue().size(), this.getLargestPoolSize(),
                this.getMaximumPoolSize(), this.getKeepAliveTime(TimeUnit.MILLISECONDS), this.isShutdown(), this.isTerminated());

        List<Tag> tags = Arrays.asList(new ImmutableTag("poolName", poolName));
        Metrics.gauge("thread_pool_active_size", tags,this.getActiveCount());

    }


    public String getPoolName() {
        return poolName;
    }


}
